/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.google.code.motoo.core;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.google.code.motoo.filter.IoFilter;
import com.google.code.motoo.filter.IoFilterChain;
import com.google.code.motoo.utils.IoSessionUtils;

/**
 * ReaderWorker的管理类，负责创建和销毁ReaderWorker类，同时提供合适的ReaderWorker类
 * 用于新channel的注册
 * @author linfengqi
 * @version 1.0  May 13, 2009
 */
public class Reader {
	
	private int corePoolSize;
	private int maximumPoolSize;
	private int threshold;
	private long keepAliveTime;
	private volatile int   poolSize;
	private ReaderWorker[] workerGroup = new ReaderWorker[0]; 
	
	public Reader() {
		this(1, 1, Integer.MAX_VALUE, Long.MAX_VALUE);
	}
	
	public Reader(int corePoolSize, int maximumPoolSize, int threshold, long keepAliveTime) {
		this.corePoolSize = corePoolSize;
		this.maximumPoolSize = maximumPoolSize;
		this.threshold = threshold;
		this.keepAliveTime = keepAliveTime;
	}
	
	public void registerChannel(SocketChannel channel, IoSession session) throws IOException {
		ReaderWorker worker = chooseWorker();
		worker.registerChannel(channel, session);
	}
	
	public ReaderWorker chooseWorker() throws IOException {
		if (poolSize < corePoolSize)
			return addWorker();

		if (overThreshold() && poolSize < maximumPoolSize)
			return addWorker();

		return minSizeWorker();
	}
	
	public void monitorSelector() throws IOException {
		if (poolSize <= corePoolSize) return;
		
		long timeout = keepAliveTime;
		for (int i = 0; i < workerGroup.length; i++) {
			ReaderWorker worker = workerGroup[i];
			if (worker.isTimeout(timeout))
				destoryWorker(i);
		}
	}
	
	public synchronized void destoryWorker(int index) throws IOException {
		ReaderWorker worker = workerGroup[index];
		
		int size = workerGroup.length;
		int numMoved = size - index - 1;
		ReaderWorker[] newWorkerGroup = new ReaderWorker[size - 1];
		System.arraycopy(workerGroup, 0, newWorkerGroup, 0, index);
		System.arraycopy(workerGroup, index + 1, newWorkerGroup, index, numMoved);
		workerGroup = newWorkerGroup;
		--poolSize;
		
		worker.destory();
	}
	
	private ReaderWorker minSizeWorker() {
		int minSize = workerGroup[0].channelCount();
		int index = 0;

		for (int i = 0; i < workerGroup.length; i++) {
			ReaderWorker worker = workerGroup[i];
			int current = worker.channelCount();
			if (current < minSize) {
				minSize = current;
				index = i;
			}
		}
		return workerGroup[index];
	}
	
	private boolean overThreshold() {
		boolean bigger = true;
		for (int i = 0; i < workerGroup.length; i++) {
			ReaderWorker worker = workerGroup[i];
			int current = worker.channelCount();
			if (current < threshold)
				bigger = false;
		}
		return bigger;
	}
	
	private synchronized ReaderWorker addWorker() throws IOException {
		ReaderWorker worker = new ReaderWorker();
		Thread thread = new Thread(worker);
		thread.setDaemon(false);
		thread.start();
		addWorkerToGroup(worker);
		++poolSize;
		return worker;
	}
	
	private void addWorkerToGroup(ReaderWorker worker) {
		int currentSize = workerGroup.length;
		ReaderWorker[] newWorkerGroup = new ReaderWorker[currentSize + 1];
		if (currentSize > 0) {
			System.arraycopy(workerGroup, 0, newWorkerGroup, 0, currentSize);  
		}
		newWorkerGroup[currentSize] = worker;
		workerGroup = newWorkerGroup;
	}

	public int getCorePoolSize() {
		return corePoolSize;
	}

	public void setCorePoolSize(int corePoolSize) {
		this.corePoolSize = corePoolSize;
	}

	public int getMaximumPoolSize() {
		return maximumPoolSize;
	}

	public void setMaximumPoolSize(int maximumPoolSize) {
		this.maximumPoolSize = maximumPoolSize;
	}

	public int getThreshold() {
		return threshold;
	}

	public void setThreshold(int threshold) {
		this.threshold = threshold;
	}

	public long getKeepAliveTime() {
		return keepAliveTime;
	}

	public void setKeepAliveTime(long keepAliveTime) {
		this.keepAliveTime = keepAliveTime;
	}

	public int getPoolSize() {
		return poolSize;
	}
	
	/**
	 * 从channel读取数据的类，将读到的数据放入IoBuffer中。每一个ReaderWorker是一个线程
	 * @author linfengqi
	 * @version 1.0  May 18, 2009
	 */
	static class ReaderWorker implements Runnable {
		
		public final static int BLOCK = 1024;
		private volatile boolean running = true;
		private volatile long lastAccessTime;
		private volatile int channelCount;
		private volatile boolean isRegister;
		private ByteBuffer fixedBuffer = ByteBuffer.allocate(BLOCK);
		public final static Log log = LogFactory.getLog(ReaderWorker.class);
		
		private Selector selector;
		
		public ReaderWorker() throws IOException {
			this(Selector.open());
		}
		
		public ReaderWorker(Selector selector) {
			this.selector = selector;
		}
		
		public synchronized void registerChannel(SocketChannel channel, IoSession session) throws IOException {
			channel.configureBlocking(false);
			isRegister = true;
			selector.wakeup();
			channel.register(selector, SelectionKey.OP_READ, session);
			log.debug("register channel from " + channel.socket().getRemoteSocketAddress());
			isRegister = false;
			synchronized (selector) {
				selector.notify();
			}
			++channelCount;
		}
		
		public void run() {
			while (running) {
				SelectionKey key = null;
				try {
					selector.select();
					Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
					while (iter.hasNext()) {
						key = iter.next();
						iter.remove();
						if (key.isValid() && key.isReadable()) {
								doRead(key);
						} 
						key = null;
					}
					if (isRegister) {
						synchronized (selector) {
							selector.wait();
						}
					}
				} catch (Exception e) {
					log.fatal("reader selector happens error!" + e);
					e.printStackTrace();
				}
			}
			
		}
		
		// 读信息
		private void doRead(SelectionKey key) {
			SocketChannel channel = (SocketChannel) key.channel();
			if (!channel.isOpen()) return;
			IoSession session = (IoSession) key.attachment();
			IoBuffer ioBuffer = session.getIoBuffer();
			IoHandler ioHandler = session.getHandler();
			session.setLastAccessTime();
			int count;
			try {
				count = channel.read(fixedBuffer);
				if (count > 0) {
					fixedBuffer.flip();
					ioBuffer.put(fixedBuffer);
					ioBuffer.flip();
					IoSessionUtils.setThreadLocalIoSession(session);
					IoFilterChain chain = (IoFilterChain) session.getAttribute(IoSession.IO_FILTER_CHAIN);
					IoFilter head = chain.getFirst();
					head.messageReceived(chain.getNextFilter(head), session, ioBuffer);
				}
				if (count < 0) {
					throw new ClientConnectionClosedException("client is closed!");
				}
				fixedBuffer.clear();
			} catch (Exception e) {
				try {
					closeConnection(key);
					ioHandler.exceptionCaught(session, e);
				} catch (Exception e1) {
					log.warn("read data happen error!", e1);
				}
			} 
		}
		
		private void closeConnection(SelectionKey key) throws IOException {
			if (key != null) {
				IoSession session = (IoSession) key.attachment();
				if (session != null) session.close();
				key.cancel();
			}
		}
		
		public int channelCount() {
			return selector.keys().size();
		}
		
		public void flushAccessTime() {
			if (channelCount() == 0)
				lastAccessTime = System.currentTimeMillis();
		}

		public boolean isTimeout(long keepAliveTime) {
			if (channelCount() != 0)
				return false;
			if ((System.currentTimeMillis() - lastAccessTime) > keepAliveTime)
				return true;
			return false;
		}
		
		public Selector selector() {
			return selector;
		}
		
		public void destory() throws IOException {
			running = false;
			selector.wakeup();
			selector.close();
		}
		
	}
	

}
